package trouger.dmt;

import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.util.ArrayList;

import org.apache.hadoop.mapred.JobConf;

class BaseMapredUtils {
	
	public static String SEPARATOR = "trouger.random.separator-THjABjmvoJgQseAKBtAsRmcKTsdIlYTe";
	public static String TERMINATOR = "trouger.terminator";
	public static String NULLFLAG = "trouger.random.nullflag-ObxsiXoXkVYiqMIsOXsTIOfmjwSBOtXC";
	
	static public int[][] getKeyColumnList(int sourceCount, JobConf conf){
		int[][] keyColumnList;
		String[] columnListsStrArray = BaseMapredUtils.getConfigStrings(conf, BaseMapredParams.KEY_COLUMN_LISTS_PARAM_NAME);
		if (columnListsStrArray == null){
			return new int[sourceCount][0];
		}
		if (columnListsStrArray.length != sourceCount){
			throw new IllegalArgumentException("the number of key-column-lists is not source-count");
		}
		keyColumnList = new int[sourceCount][];
		int keyColumnNumber = -1;
		for (int i = 0; i < sourceCount; i++){
			String[] indices = columnListsStrArray[i].split("\\s");
			if (keyColumnNumber < 0){
				keyColumnNumber = indices.length;
			}else if (keyColumnNumber != indices.length){
				throw new IllegalArgumentException("number of key-columns not consistent among sources");
			}
			keyColumnList[i] = new int[indices.length];
			for (int j = 0; j < indices.length; j++){
				keyColumnList[i][j] = Integer.parseInt(indices[j]);
			}
		}
		if (keyColumnNumber == 0){
			throw new IllegalArgumentException("key-columns cannot be empty");
		}
		return keyColumnList;
	}
	
	@Deprecated
	static public int[][] getValueColumnList(int sourceCount, JobConf conf){
		int[][] valueColumnList; 
		String[] columnListsStrArray = BaseMapredUtils.getConfigStrings(conf, BaseMapredParams.VALUE_COLUMN_LISTS_PARAM_NAME);
		if (columnListsStrArray == null){
			throw new IllegalArgumentException("value-column-lists are not provided");
		}else if (columnListsStrArray.length != sourceCount){
			throw new IllegalArgumentException("the number of value-column-lists is not source-count");
		}
		valueColumnList = new int[sourceCount][];
		for (int i = 0; i < sourceCount; i++){
			if (columnListsStrArray[i] != null){
				String[] indices = columnListsStrArray[i].split("\\s");
				valueColumnList[i] = new int[indices.length];
				for (int j = 0; j < indices.length; j++){
					valueColumnList[i][j] = Integer.parseInt(indices[j]);
				}
			}else{
				valueColumnList[i] = new int[0];
			}
		}
		return valueColumnList;
	}
	
	/**
	 * Get a least-redundancy value column list.
	 * Columns that appear in keyColumnList will not be put into the value column list.
	 * @param sourceCount the source count
	 * @param conf the JobConf instance
	 * @param keyColumnList the key column list, generally returned by a previous call to {@link #getKeyColumnList}
	 * @return the value column list
	 */
	static public int[][] getValueColumnListEx(int sourceCount, JobConf conf, int[][] keyColumnList){
		int[][] valueColumnList; 
		String[] columnListsStrArray = BaseMapredUtils.getConfigStrings(conf, BaseMapredParams.VALUE_COLUMN_LISTS_PARAM_NAME);
		if (columnListsStrArray == null){
			throw new IllegalArgumentException("value-column-lists are not provided");
		}else if (columnListsStrArray.length != sourceCount){
			throw new IllegalArgumentException("the number of value-column-lists is not source-count");
		}
		valueColumnList = new int[sourceCount][];
		for (int i = 0; i < sourceCount; i++){
			if (columnListsStrArray[i] != null){
				String[] indices = columnListsStrArray[i].split("\\s");
				ArrayList<Integer> arrayList = new ArrayList<Integer>();
				for (int j = 0; j < indices.length; j++){
					int index = Integer.parseInt(indices[j]);
					boolean flag = true;
					for (int k = 0; k < keyColumnList[i].length; k++){
						if (keyColumnList[i][k] == index) {
							flag = false;
							break;
						}
					}
					if (flag) arrayList.add(index);
				}
				valueColumnList[i] = new int[arrayList.size()];
				for (int j = 0; j < arrayList.size(); j++){
					valueColumnList[i][j] = arrayList.get(j);
				}
			}else{
				valueColumnList[i] = new int[0];
			}
		}
		return valueColumnList;
	}
	
	static public class FieldIndicesStringParseResult{
		private FieldIndexPair[] fieldIndices;
		private int sourceCount;
		public FieldIndexPair[] getFieldIndices() {
			return fieldIndices;
		}
		public int getGuessedSourceCount() {
			return sourceCount;
		}
		public FieldIndicesStringParseResult(FieldIndexPair[] fieldIndices, int sourceCount){
			this.fieldIndices = fieldIndices;
			this.sourceCount = sourceCount;
		}
	}
	
	static public FieldIndicesStringParseResult parseFieldIndicesString(String fieldIndicesString){
		String[] seg = fieldIndicesString.split("[\\s]*[\\s,][\\s]*");
		FieldIndexPair[] fieldIndices = new FieldIndexPair[seg.length];
		int sourceCount = 0;
		for (int i = 0; i < seg.length; i++){
			String[] part = seg[i].split("\\.");
			if (part.length != 2){
				throw new IllegalArgumentException("wrong format of field-indices string: " + fieldIndicesString);
			}
			int sourceIndex = Integer.parseInt(part[0]);
			int columnIndex = Integer.parseInt(part[1]);
			if (sourceIndex <= 0){
				throw new IllegalArgumentException("source index must be positive");
			}
			fieldIndices[i] = new FieldIndexPair(sourceIndex, columnIndex);
			if (sourceIndex > sourceCount){
				sourceCount = sourceIndex;
			}
		}
		return new FieldIndicesStringParseResult(fieldIndices, sourceCount);
	}
	
	public static void setConfigString(JobConf conf, String name, String value){
		try {
			conf.set(name, URLEncoder.encode(value, "utf-8"));
		} catch (UnsupportedEncodingException e) {
			throw new UnsupportedOperationException("utf-8 encoding not found in the system");
		}
	}
	
	public static String getConfigString(JobConf conf, String name){
		String s = conf.get(name);
		if (s == null) return null;
		try {
			return URLDecoder.decode(s, "utf-8");
		} catch (UnsupportedEncodingException e) {
			throw new UnsupportedOperationException("utf-8 encoding not found in the system");
		}
	}
	
	public static void setConfigStrings(JobConf conf, String name, String[] strings) {
		StringBuffer strbuf = new StringBuffer();
		for (String s : strings){
			if (s != null){
				strbuf.append(s);
			}else{
				strbuf.append(NULLFLAG);
			}
			strbuf.append(SEPARATOR);
		}
		strbuf.append(TERMINATOR);
		try {
			conf.set(name, URLEncoder.encode(strbuf.toString(), "utf-8"));
		} catch (UnsupportedEncodingException e) {
			throw new UnsupportedOperationException("utf-8 encoding not found in the system");
		}
	}
	public static void setConfigStrings(JobConf conf, String name, ArrayList<? extends Object> arrayList){
		int i = 0;
		String[] strings = new String[arrayList.size()];
		for (Object o : arrayList){
			if (o != null){
				strings[i++] = o.toString();
			}else{
				strings[i++] = null;
			}
		}
		setConfigStrings(conf, name, strings);
	}
	
	public static String[] getConfigStrings(JobConf conf, String name){
		String s = conf.get(name);
		if (s == null){
			return null;
		}else{
			try {
				s = URLDecoder.decode(s, "utf-8");
			} catch (UnsupportedEncodingException e) {
				throw new UnsupportedOperationException("utf-8 encoding not found in the system");
			}
			String[] strings = s.split(SEPARATOR);
			int n;
			String[] ret = new String[n = strings.length - 1];
			for (int i = 0; i < n; i++){
				ret[i] = strings[i];
				if (ret[i].equals(NULLFLAG)){
					ret[i] = null;
				}
			}
			return ret;
		}
	}
	
	public static FieldIndexPair[] getResultFieldIndices(JobConf conf, int sourceCount){
		String resultFieldsIndicesStr = conf.get(BaseMapredParams.RESULT_FIELDS_INDICES_PARAM_NAME);
		if (resultFieldsIndicesStr == null){
			return null;
		}else{
			String[] fieldsList = resultFieldsIndicesStr.split("\\s");
			int n = fieldsList.length;
			FieldIndexPair[] resultFieldsIndex = new FieldIndexPair[n];
			for (int i = 0; i < n; i++){
				String[] part = fieldsList[i].split("\\.");
				if (part.length != 2){
					throw new IllegalArgumentException("wrong format of result-fields-indices");
				}else{
					int j = Integer.parseInt(part[0]);
					int k = Integer.parseInt(part[1]);
					if (j > sourceCount){
						throw new IllegalArgumentException("source index exceeds source count");
					}
					resultFieldsIndex[i] = new FieldIndexPair(j, k);
				}
			}
			return resultFieldsIndex;
		}
	}
	
	public static String[] plainTextSplit(String s, String del){
		ArrayList<String> ret = new ArrayList<String>();
		int p = 0;
		int q;
		while ((q = s.indexOf(del, p)) >= 0){
			ret.add(s.substring(p, q));
			p = q + del.length();
		}
		ret.add(s.substring(p));
		String[] reta = new String[ret.size()];
		return ret.toArray(reta);
	}
}
